package main

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"time"

	"natproxy/pkg/common"
	"natproxy/pkg/tunnel"

	"github.com/sirupsen/logrus"
)

type Server struct {
	clients            *sync.Map
	clientsMux         sync.RWMutex
	OnNewConnection    func(net.Conn) bool
	OnRequest          func(*common.HTTPRequest, string) bool
	OnConnectionClosed func(net.Conn)
	minClientsPerID    int
	maxClientsPerID    int
	activeConnections  *sync.Map
	healthCheck        *time.Ticker
	stateFile          string
	clientStates       sync.Map
	loggers            map[string]*common.LogManager
	logger             *common.LogManager
	connectionTimeout  time.Duration // 连接超时时间
	heartbeatTimeout   time.Duration // 心跳超时时间
	lastHeartbeats     *sync.Map     // 记录每个客户端的最后心跳时间
	config             *Config       // 添加配置字段
	maxConnections     int32         // 最大连接数
	currentConnections int32         // 当前连接数
	activeClients      sync.Map      // 存储活跃的客户端ID，确保唯一连接
}

type ClientHandler struct {
	tunnel      *tunnel.Tunnel
	respChans   map[string]chan *common.TunnelMessage
	respMux     sync.RWMutex
	done        chan struct{}
	maxRequests int
	semaphore   chan struct{}
	localPort   int // 添加本地端口配置
}

const (
	MaxLogLines        = 5000                  // 单个日志文件最大行数
	LogsRetention      = 7                     // 日志保留天数
	maxRetries         = 3                     // 最大重试次数
	retryDelay         = 10 * time.Millisecond // 重试等待时间
	clientStateFile    = "client_state.json"   // 客户端状态文件
	stateCheckInterval = time.Minute           // 状态检查间隔
	MaxChunkSize       = 10 * 1024 * 1024      // 10MB 分片大小
	ChunkTimeout       = 30 * time.Second      // 分片请求超时时间
	FileLogMinSize     = 1024 * 1024           // 1MB 以上的文件传输才记录日志
)

type LogManager struct {
	currentFile    *os.File
	currentLines   int
	currentLogDate string
	logDir         string
	mu             sync.Mutex
}

func NewLogManager(logDir string) (*LogManager, error) {
	if err := os.MkdirAll(logDir, 0755); err != nil {
		return nil, fmt.Errorf("创建日志目录失败: %v", err)
	}

	manager := &LogManager{
		logDir: logDir,
	}

	go manager.cleanupRoutine()

	if err := manager.rotateIfNeeded(); err != nil {
		return nil, err
	}

	return manager, nil
}

func (lm *LogManager) Write(p []byte) (n int, err error) {
	lm.mu.Lock()
	defer lm.mu.Unlock()

	if lm.currentLines%100 == 0 {
		if err := lm.rotateIfNeeded(); err != nil {
			return 0, err
		}
	}

	n, err = lm.currentFile.Write(p)
	if err != nil {
		return n, err
	}

	lm.currentLines += bytes.Count(p, []byte{'\n'})
	return n, nil
}

func (lm *LogManager) rotateIfNeeded() error {
	today := time.Now().Format("2006-01-02")

	if lm.currentFile == nil || today != lm.currentLogDate || lm.currentLines >= MaxLogLines {
		if lm.currentFile != nil && lm.currentLines >= MaxLogLines {
			lm.currentFile.Close()

			timestamp := time.Now().Format("150405")
			newFileName := fmt.Sprintf("server_%s.log", timestamp)
			newPath := filepath.Join(lm.logDir, today, newFileName)

			if err := os.MkdirAll(filepath.Join(lm.logDir, today), 0755); err != nil {
				return err
			}

			oldPath := filepath.Join(lm.logDir, today, "server.log")
			if err := os.Rename(oldPath, newPath); err != nil {
				return fmt.Errorf("移动日志文件失败: %v", err)
			}
		}

		todayDir := filepath.Join(lm.logDir, today)
		if err := os.MkdirAll(todayDir, 0755); err != nil {
			return fmt.Errorf("创建日期目录失败: %v", err)
		}

		newFile, err := os.OpenFile(
			filepath.Join(todayDir, "server.log"),
			os.O_CREATE|os.O_APPEND|os.O_WRONLY,
			0644,
		)
		if err != nil {
			return fmt.Errorf("创建新日志文件失败: %v", err)
		}

		if lm.currentFile != nil {
			lm.currentFile.Close()
		}

		lm.currentFile = newFile
		lm.currentLogDate = today
		lm.currentLines = 0

		if lines, err := lm.countFileLines(newFile.Name()); err == nil {
			lm.currentLines = lines
		}
	}

	return nil
}

func (lm *LogManager) cleanupRoutine() {
	ticker := time.NewTicker(24 * time.Hour)
	defer ticker.Stop()

	for range ticker.C {
		lm.cleanup()
	}
}

func (lm *LogManager) cleanup() {
	cutoff := time.Now().AddDate(0, 0, -LogsRetention)

	entries, err := os.ReadDir(lm.logDir)
	if err != nil {
		log.Printf("读取日志目录失败: %v", err)
		return
	}

	for _, entry := range entries {
		if !entry.IsDir() {
			continue
		}

		dirDate, err := time.Parse("2006-01-02", entry.Name())
		if err != nil {
			continue
		}

		if dirDate.Before(cutoff) {
			dirPath := filepath.Join(lm.logDir, entry.Name())
			if err := os.RemoveAll(dirPath); err != nil {
				log.Printf("删除旧日志目录失败 %s: %v", dirPath, err)
			} else {
				log.Printf("已删除旧日志目录: %s", dirPath)
			}
		}
	}
}

func (lm *LogManager) countFileLines(filename string) (int, error) {
	file, err := os.Open(filename)
	if err != nil {
		return 0, err
	}
	defer file.Close()

	scanner := bufio.NewScanner(file)
	lineCount := 0
	for scanner.Scan() {
		lineCount++
	}
	return lineCount, scanner.Err()
}

type ClientState struct {
	ClientID  string    `json:"client_id"`
	LastSeen  time.Time `json:"last_seen"`
	IsActive  bool      `json:"is_active"`
	ConnCount int       `json:"conn_count"`
}

func NewServer(config *Config) (*Server, error) {
	s := &Server{
		clients:           new(sync.Map),
		minClientsPerID:   config.Server.MinClientsPerID,       // 从配置读取
		maxClientsPerID:   config.Server.MaxClientsPerID,       // 从配置读取
		maxConnections:    int32(config.Server.MaxConnections), // 从配置读取
		activeConnections: new(sync.Map),
		healthCheck:       time.NewTicker(30 * time.Second),
		stateFile:         "data/" + clientStateFile,
		clientStates:      sync.Map{},
		loggers:           make(map[string]*common.LogManager),
		connectionTimeout: 5 * time.Minute,  // 5分钟连接超时
		heartbeatTimeout:  90 * time.Second, // 90秒心跳超时
		lastHeartbeats:    new(sync.Map),
		config:            config, // 保存配置
	}

	// 初始化所有日志管理器
	logTypes := []string{
		"main", // 添加 main 类型
		common.LogTypeError,
		common.LogTypeAccess,
		common.LogTypeTransfer,
		common.LogTypeSecurity,
	}

	for _, logType := range logTypes {
		logger, err := common.NewLogManager(&config.Server.LogConfig, logType)
		if err != nil {
			return nil, fmt.Errorf("初始化%s日志失败: %v", logType, err)
		}
		if logType == "main" {
			s.logger = logger // 设置主日志管理器
		} else {
			s.loggers[logType] = logger
		}
	}

	os.MkdirAll("data", 0755)
	s.loadClientStates()

	go s.stateManagementRoutine()
	go s.healthCheckRoutine()

	// 启动连接监控
	go s.monitorConnections()

	return s, nil
}

func NewClientHandler(t *tunnel.Tunnel, maxRequests int, localPort int) *ClientHandler {
	if maxRequests <= 0 {
		maxRequests = 100
	}
	return &ClientHandler{
		tunnel:      t,
		respChans:   make(map[string]chan *common.TunnelMessage),
		done:        make(chan struct{}),
		maxRequests: maxRequests,
		semaphore:   make(chan struct{}, maxRequests),
		localPort:   localPort,
	}
}

func (h *ClientHandler) addResponseChan(requestID string) chan *common.TunnelMessage {
	h.respMux.Lock()
	defer h.respMux.Unlock()

	ch := make(chan *common.TunnelMessage, 1)
	h.respChans[requestID] = ch
	return ch
}

func (h *ClientHandler) removeResponseChan(requestID string) {
	h.respMux.Lock()
	defer h.respMux.Unlock()

	if ch, exists := h.respChans[requestID]; exists {
		close(ch)
		delete(h.respChans, requestID)
	}
}

func (h *ClientHandler) getResponseChan(requestID string) (chan *common.TunnelMessage, bool) {
	h.respMux.RLock()
	defer h.respMux.RUnlock()

	ch, exists := h.respChans[requestID]
	return ch, exists
}

func (h *ClientHandler) handleRequest(msg *common.TunnelMessage) {
	h.semaphore <- struct{}{}        // 获取信号量
	defer func() { <-h.semaphore }() // 释放信号量

	var httpReq common.HTTPRequest
	if err := json.Unmarshal(msg.Payload, &httpReq); err != nil {
		log.Printf("解析请求失败: %v", err)
		h.sendErrorResponse(msg.RequestID, fmt.Sprintf("解析请求失败: %v", err))
		return
	}

	// 转发请求到本地服务
	localAddr := fmt.Sprintf("http://localhost:%d%s", h.localPort, httpReq.Path)
	req, err := http.NewRequest(httpReq.Method, localAddr, bytes.NewReader(httpReq.Body))
	if err != nil {
		log.Printf("创建本地请求失败: %v", err)
		h.sendErrorResponse(msg.RequestID, fmt.Sprintf("创建本地请求失败: %v", err))
		return
	}

	// 设置请求头
	for k, v := range httpReq.Headers {
		req.Header.Set(k, v)
	}

	// 设置超时
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	req = req.WithContext(ctx)

	// 发送请求
	client := &http.Client{
		Timeout: 30 * time.Second,
	}
	resp, err := client.Do(req)
	if err != nil {
		log.Printf("发送本地请求失败: %v", err)
		h.sendErrorResponse(msg.RequestID, fmt.Sprintf("发送本地请求失败: %v", err))
		return
	}
	defer resp.Body.Close()

	// 读取响应
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		log.Printf("读取响应失败: %v", err)
		h.sendErrorResponse(msg.RequestID, fmt.Sprintf("读取响应失败: %v", err))
		return
	}

	// 构造响应消息
	httpResp := common.HTTPResponse{
		StatusCode: resp.StatusCode,
		Headers:    make(map[string]string),
		Body:       body,
	}

	// 复制响应头
	for k, v := range resp.Header {
		if len(v) > 0 {
			httpResp.Headers[k] = v[0]
		}
	}

	// 序列化响应
	respBytes, err := json.Marshal(httpResp)
	if err != nil {
		log.Printf("序列化响应失败: %v", err)
		h.sendErrorResponse(msg.RequestID, fmt.Sprintf("序列化响应失败: %v", err))
		return
	}

	// 发送响应
	err = h.tunnel.SendMessage(&common.TunnelMessage{
		Type:      common.TypeHTTPResponse,
		RequestID: msg.RequestID,
		Payload:   respBytes,
	})

	if err != nil {
		log.Printf("发送响应失败: %v", err)
	}
}

// 添加错误响应方法
func (h *ClientHandler) sendErrorResponse(requestID string, errMsg string) {
	resp := common.HTTPResponse{
		StatusCode: http.StatusInternalServerError,
		Headers:    make(map[string]string),
		Body:       []byte(errMsg),
	}

	respBytes, _ := json.Marshal(resp)
	h.tunnel.SendMessage(&common.TunnelMessage{
		Type:      common.TypeHTTPResponse,
		RequestID: requestID,
		Payload:   respBytes,
	})
}

func (s *Server) handleHTTPRequest(w http.ResponseWriter, r *http.Request) {
	// 从URL路径中提取客户端ID
	parts := strings.SplitN(r.URL.Path[1:], "/", 2)
	if len(parts) < 2 {
		fmt.Printf("[错误] 无效的请求路径: %s\n", r.URL.Path)
		http.Error(w, "Invalid path", http.StatusBadRequest)
		return
	}

	clientID := parts[0]
	actualPath := "/" + parts[1]

	fmt.Printf("[请求] 收到外部请求: %s %s (客户端: %s)\n", r.Method, actualPath, clientID)

	// 获取客户端处理器
	handlerObj, exists := s.clients.Load(clientID)
	if !exists {
		fmt.Printf("[错误] 未找到客户端: %s\n", clientID)
		http.Error(w, "Client not found", http.StatusNotFound)
		return
	}
	handler := handlerObj.(*ClientHandler)

	// 读取请求体
	body, err := io.ReadAll(r.Body)
	if err != nil {
		fmt.Printf("[错误] 读取请求体失败: %v\n", err)
		http.Error(w, "Failed to read request body", http.StatusInternalServerError)
		return
	}

	// 构造隧道请求
	httpReq := &common.HTTPRequest{
		Method:  r.Method,
		Path:    actualPath, // 使用处理后的路径
		Headers: make(map[string]string),
		Body:    body,
	}

	// 复制请求头
	for k, v := range r.Header {
		if len(v) > 0 {
			httpReq.Headers[k] = v[0]
		}
	}

	// 转发请求
	resp, err := handler.forwardRequest(httpReq)
	if err != nil {
		fmt.Printf("[错误] 转发请求失败: %v\n", err)
		http.Error(w, "Failed to forward request", http.StatusInternalServerError)
		return
	}

	// 打印响应信息
	fmt.Printf("[响应] 客户端 %s 返回: %d\n", clientID, resp.StatusCode)

	// 设置响应头
	for k, v := range resp.Headers {
		w.Header().Set(k, v)
	}

	// 写入响应
	w.WriteHeader(resp.StatusCode)
	w.Write(resp.Body)
}

func readRequestBody(r *http.Request, contentLength int64) ([]byte, error) {
	if contentLength <= 0 {
		return io.ReadAll(r.Body)
	}

	if contentLength > MaxChunkSize {
		return readLargeRequestBody(r, contentLength)
	}

	body := make([]byte, contentLength)
	_, err := io.ReadFull(r.Body, body)
	return body, err
}

func readLargeRequestBody(r *http.Request, contentLength int64) ([]byte, error) {
	var buf bytes.Buffer
	if _, err := io.CopyN(&buf, r.Body, contentLength); err != nil {
		return nil, err
	}
	return buf.Bytes(), nil
}

func MinInt64(a, b int64) int64 {
	if a < b {
		return a
	}
	return b
}

func (s *Server) forwardRequest(msg *common.TunnelMessage, timeout time.Duration) (*common.HTTPResponse, error) {
	handler, exists := s.clients.Load(msg.Target)
	if !exists {
		return nil, fmt.Errorf("客户端 %s 未连接", msg.Target)
	}

	respChan := handler.(*ClientHandler).addResponseChan(msg.RequestID)
	defer handler.(*ClientHandler).removeResponseChan(msg.RequestID)

	if err := handler.(*ClientHandler).tunnel.SendMessage(msg); err != nil {
		return nil, fmt.Errorf("发送请求失败: %v", err)
	}

	select {
	case resp := <-respChan:
		var httpResp common.HTTPResponse
		if err := json.Unmarshal(resp.Payload, &httpResp); err != nil {
			return nil, fmt.Errorf("解析响应失败: %v", err)
		}
		return &httpResp, nil
	case <-time.After(timeout):
		return nil, fmt.Errorf("请求超时")
	}
}

type Config struct {
	Server struct {
		TCPPort         int              `json:"tcp_port"`
		HTTPPort        int              `json:"http_port"`
		LocalPort       int              `json:"local_port"`
		MaxConnections  int              `json:"max_connections"`
		MaxClientsPerID int              `json:"max_clients_per_id"`
		MinClientsPerID int              `json:"min_clients_per_id"`
		LogConfig       common.LogConfig `json:"log_config"`
	} `json:"server"`
}

func loadConfig() (*Config, error) {
	// 尝试多个可能的配置文件位置
	possiblePaths := []string{
		"config/server.json",       // 当前目录下的 config 目录
		"../config/server.json",    // 上级目录的 config 目录
		"../../config/server.json", // 项目根目录
		os.Getenv("CONFIG_PATH"),   // 环境变量指定的路径
	}

	var configFile []byte
	var err error
	var loadedPath string

	// 尝试所有可能的路径
	for _, path := range possiblePaths {
		if path == "" {
			continue
		}
		configFile, err = os.ReadFile(path)
		if err == nil {
			loadedPath = path
			break
		}
	}

	if err != nil {
		return nil, fmt.Errorf("无法找到配置文件，尝试过以下路径: %v", possiblePaths)
	}

	var config Config
	if err := json.Unmarshal(configFile, &config); err != nil {
		return nil, fmt.Errorf("解析配置文件 %s 失败: %v", loadedPath, err)
	}

	log.Printf("成功加载配置文件: %s", loadedPath)
	return &config, nil
}

func main() {
	logManager, err := NewLogManager("logs")
	if err != nil {
		log.Fatal("创建日志管理器失败:", err)
	}

	log := logrus.New()
	log.SetOutput(io.MultiWriter(os.Stdout, logManager))
	log.SetFormatter(&logrus.JSONFormatter{})

	config, err := loadConfig()
	if err != nil {
		log.Fatal("加载配置文件失败:", err)
	}

	server, err := NewServer(config)
	if err != nil {
		log.Fatal("创建服务器失败:", err)
	}

	securityConfig := &common.SecurityConfig{
		MaxRequestsPerSecond: 1000,
		BlockDuration:        time.Minute * 5,
		MaxConnPerIP:         10,
		MaxRequestSize:       1024 * 1024 * 10,
		WhiteList:            []string{"127.0.0.1"},
		BlackList:            []string{},
	}

	securityManager := common.NewSecurityManager(securityConfig)

	server.OnNewConnection = func(conn net.Conn) bool {
		ip := getIPFromConn(conn)

		if !securityManager.IsIPAllowed(ip) {
			log.Printf("拒绝来自 %s 的连接: IP被封禁", ip)
			return false
		}

		if !securityManager.IncrementConnCount(ip) {
			log.Printf("拒绝来自 %s 的连接: 超过最大连接数限制", ip)
			return false
		}

		return true
	}

	server.OnRequest = func(req *common.HTTPRequest, clientIP string) bool {
		if !securityManager.IncrementRequestCount(clientIP) {
			log.Printf("拒绝来自 %s 的请求: 请求频率过高", clientIP)
			return false
		}

		if int64(len(req.Body)) > securityConfig.MaxRequestSize {
			log.Printf("拒绝来自 %s 的请求: 请求体积过大", clientIP)
			return false
		}

		return true
	}

	server.OnConnectionClosed = func(conn net.Conn) {
		ip := getIPFromConn(conn)
		securityManager.DecrementConnCount(ip)
	}

	tcpAddr := fmt.Sprintf(":%d", config.Server.TCPPort)
	listener, err := net.Listen("tcp", tcpAddr)
	if err != nil {
		log.Fatal("TCP监听失败:", err)
	}
	log.Printf("TCP服务已启动，正在监听端口 %d", config.Server.TCPPort)

	// 使用 listener 变量
	go func() {
		for {
			conn, err := listener.Accept()
			if err != nil {
				log.Printf("接受连接失败: %v", err)
				continue
			}
			go server.handleTunnelConnection(conn)
		}
	}()

	httpAddr := fmt.Sprintf(":%d", config.Server.HTTPPort)
	go func() {
		if err := http.ListenAndServe(httpAddr, http.HandlerFunc(server.handleHTTPRequest)); err != nil {
			log.Printf("HTTP服务启动失败: %v", err)
		}
	}()

	log.Printf("服务端启动成功 - TCP端口: %d, HTTP端口: %d",
		config.Server.TCPPort, config.Server.HTTPPort)

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	go func() {
		<-sigChan
		log.Println("正在关闭服务...")
		cancel()
	}()

	<-ctx.Done()

	server.saveClientStates()
	server.closeAllConnections()
}

func getIPFromConn(conn net.Conn) string {
	addr := conn.RemoteAddr().String()
	ip, _, _ := net.SplitHostPort(addr)
	return ip
}

func (s *Server) healthCheckRoutine() {
	for range s.healthCheck.C {
		s.checkClientConnections()
	}
}

func (s *Server) checkClientConnections() {
	s.clientStates.Range(func(key, value interface{}) bool {
		clientID := key.(string)
		state := value.(*ClientState)

		if state.ConnCount < s.minClientsPerID {
			s.logger.Log(common.LogWarn, "[监控] 检测到客户端 %s 连接数不足: %d/%d",
				clientID, state.ConnCount, s.minClientsPerID)
		}
		return true
	})
}

func (s *Server) loadClientStates() {
	data, err := os.ReadFile(s.stateFile)
	if err != nil {
		if !os.IsNotExist(err) {
			log.Printf("读取客户端状态文件失败: %v", err)
		}
		return
	}

	var states []ClientState
	if err := json.Unmarshal(data, &states); err != nil {
		log.Printf("解析客户端状态失败: %v", err)
		return
	}

	for _, state := range states {
		if state.IsActive && time.Since(state.LastSeen) < time.Hour {
			s.clientStates.Store(state.ClientID, &state)
		}
	}
}

func (s *Server) saveClientStates() {
	var states []ClientState
	s.clientStates.Range(func(key, value interface{}) bool {
		if state, ok := value.(*ClientState); ok {
			states = append(states, *state)
		}
		return true
	})

	data, err := json.MarshalIndent(states, "", "  ")
	if err != nil {
		log.Printf("序列化客户端状态失败: %v", err)
		return
	}

	if err := os.WriteFile(s.stateFile, data, 0644); err != nil {
		log.Printf("保存客户端状态失败: %v", err)
	}
}

func (s *Server) stateManagementRoutine() {
	ticker := time.NewTicker(stateCheckInterval)
	defer ticker.Stop()

	for range ticker.C {
		s.checkAndRecoverConnections()
		s.saveClientStates()
	}
}

func (s *Server) checkAndRecoverConnections() {
	s.clientStates.Range(func(key, value interface{}) bool {
		clientID := key.(string)
		state := value.(*ClientState)

		if state.IsActive {
			count, _ := s.GetClientConnections(clientID)
			if count < uint32(s.minClientsPerID) {
				s.logger.Log(common.LogInfo, "[监控] 客户端 %s 连接数不足 (%d/%d), 等待客户端重连...",
					clientID, count, s.minClientsPerID)

				// 清理可能的僵尸连接
				if handler, exists := s.clients.Load(clientID); exists {
					handler.(*ClientHandler).tunnel.Close()
					s.clients.Delete(clientID)
				}
				s.lastHeartbeats.Delete(clientID)
			}
		} else if time.Since(state.LastSeen) > time.Hour*24 {
			s.logger.Log(common.LogInfo, "[监控] 清理长时间未活动的客户端: %s", clientID)
			s.clientStates.Delete(clientID)
		}
		return true
	})
}

func (s *Server) closeAllConnections() {
	s.clients.Range(func(key, value interface{}) bool {
		clientID := key.(string)
		handler := value.(*ClientHandler)
		log.Printf("正在关闭客户端 %s 的连接", clientID)
		handler.tunnel.Close()
		return true
	})
}

func generateRequestID() string {
	return fmt.Sprintf("%d", time.Now().UnixNano())
}

func (s *Server) handleTunnelConnection(conn net.Conn) {
	remoteAddr := conn.RemoteAddr().String()
	fmt.Printf("[连接] 新的连接请求来自: %s\n", remoteAddr)

	defer func() {
		conn.Close()
		fmt.Printf("[连接] 连接已关闭: %s\n", remoteAddr)
	}()

	t := tunnel.NewTunnel(conn)
	defer t.Close()

	// 读取注册消息
	msg, err := t.ReadMessage()
	if err != nil {
		fmt.Printf("[错误] 读取注册消息失败: %v\n", err)
		return
	}

	if msg.Type != common.TypeRegister {
		fmt.Printf("[错误] 首条消息不是注册消息\n")
		return
	}

	clientID := msg.Target
	fmt.Printf("[连接] 客户端 %s 开始注册\n", clientID)

	// 创建客户端处理器
	handler := &ClientHandler{
		tunnel:    t,
		respChans: make(map[string]chan *common.TunnelMessage),
		done:      make(chan struct{}),
	}

	// 注册客户端
	s.clients.Store(clientID, handler)
	s.lastHeartbeats.Store(clientID, time.Now())
	s.activeClients.Store(clientID, true)

	// 启动消息处理循环
	for {
		msg, err := t.ReadMessage()
		if err != nil {
			fmt.Printf("[错误] 读取消息失败: %v\n", err)
			break
		}

		switch msg.Type {
		case common.TypeHeartbeat:
			s.handleHeartbeat(clientID)
			fmt.Printf("[心跳] 收到客户端 %s 心跳\n", clientID)

		case common.TypeHTTPResponse:
			fmt.Printf("[响应] 收到客户端 %s 响应\n", clientID)
			handler.handleResponse(msg)

		default:
			fmt.Printf("[警告] 收到未知类型的消息: %v\n", msg.Type)
		}
	}

	// 清理客户端资源
	s.cleanupClient(clientID)
}

func (s *Server) updateClientState(clientID string, isConnected bool) {
	value, exists := s.clientStates.Load(clientID)
	var state *ClientState
	if !exists {
		state = &ClientState{
			ClientID: clientID,
			IsActive: isConnected,
		}
	} else {
		state = value.(*ClientState)
		state.IsActive = isConnected
	}

	state.LastSeen = time.Now()
	if isConnected {
		state.ConnCount++
	} else {
		state.ConnCount--
		if state.ConnCount < 0 {
			state.ConnCount = 0
		}
	}

	s.clientStates.Store(clientID, state)
}

func (s *Server) GetClientConnections(clientID string) (uint32, bool) {
	val, ok := s.activeConnections.Load(clientID)
	if !ok {
		return 0, false
	}

	count, ok := val.(uint32)
	if !ok {
		return 0, false
	}

	return count, true
}

func (s *Server) handleLargeFile(w http.ResponseWriter, r *http.Request, clientID string, clientHandler *ClientHandler) {
	contentLength := r.ContentLength
	if contentLength <= 0 {
		http.Error(w, "Content length required", http.StatusBadRequest)
		return
	}

	// 计算需要的分片数
	chunks := (contentLength + MaxChunkSize - 1) / MaxChunkSize
	responses := make([][]byte, chunks)

	// 创建错误通道
	errChan := make(chan error, 1)

	// 并发请求分片
	var wg sync.WaitGroup
	for i := int64(0); i < chunks; i++ {
		wg.Add(1)
		go func(chunkIndex int64) {
			defer wg.Done()

			offset := chunkIndex * MaxChunkSize
			size := MinInt64(MaxChunkSize, contentLength-offset)

			// 构造分片请求
			chunkReq := &common.FileChunkRequest{
				RequestID: generateRequestID(),
				FilePath:  r.URL.Path,
				Offset:    offset,
				Size:      size,
			}

			// 序列化请求
			reqBytes, err := json.Marshal(chunkReq)
			if err != nil {
				errChan <- fmt.Errorf("序列化分片请求失败: %v", err)
				return
			}

			// 发送分片请求
			msg := &common.TunnelMessage{
				Type:      common.TypeFileChunkRequest,
				RequestID: chunkReq.RequestID,
				Target:    clientID,
				Payload:   reqBytes,
			}

			// 等待响应
			respChan := clientHandler.addResponseChan(chunkReq.RequestID)
			defer clientHandler.removeResponseChan(chunkReq.RequestID)

			if err := clientHandler.tunnel.SendMessage(msg); err != nil {
				errChan <- fmt.Errorf("发送分片请求失败: %v", err)
				return
			}

			select {
			case resp := <-respChan:
				var chunkResp common.FileChunkResponse
				if err := json.Unmarshal(resp.Payload, &chunkResp); err != nil {
					errChan <- fmt.Errorf("解析分片响应失败: %v", err)
					return
				}

				if chunkResp.Error != "" {
					errChan <- fmt.Errorf("分片请求失败: %s", chunkResp.Error)
					return
				}

				responses[chunkIndex] = chunkResp.Data

			case <-time.After(ChunkTimeout):
				errChan <- fmt.Errorf("分片请求超时")
				return
			}
		}(i)
	}

	// 等待所有分片完成或出错
	go func() {
		wg.Wait()
		close(errChan)
	}()

	// 检查是否有错误
	if err := <-errChan; err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	// 合并所有分片
	for _, chunk := range responses {
		w.Write(chunk)
	}
}

func (s *Server) monitorConnections() {
	ticker := time.NewTicker(10 * time.Second) // 更频繁地检查
	defer ticker.Stop()

	for range ticker.C {
		s.clients.Range(func(key, value interface{}) bool {
			clientID := key.(string)
			lastHeartbeat, exists := s.lastHeartbeats.Load(clientID)
			if !exists {
				s.lastHeartbeats.Store(clientID, time.Now())
				return true
			}

			timeSinceLastHeartbeat := time.Since(lastHeartbeat.(time.Time))
			if timeSinceLastHeartbeat > s.heartbeatTimeout {
				fmt.Printf("[监控] 客户端 %s 心跳超时 (最后心跳: %v 前)\n",
					clientID, timeSinceLastHeartbeat.Round(time.Second))

				// 获取客户端处理器并关闭连接
				if handler, ok := value.(*ClientHandler); ok {
					handler.tunnel.Close()
				}
				s.clients.Delete(clientID)
				s.lastHeartbeats.Delete(clientID)
				s.activeClients.Delete(clientID)
				s.updateClientState(clientID, false)
			}
			return true
		})
	}
}

// 处理心跳消息
func (s *Server) handleHeartbeat(clientID string) {
	s.lastHeartbeats.Store(clientID, time.Now())
	fmt.Printf("[心跳] 客户端 %s 心跳更新\n", clientID) // 添加控制台输出
}

func (s *Server) _getClientIDs() []string {
	ids := make([]string, 0)
	s.clients.Range(func(key, value interface{}) bool {
		ids = append(ids, key.(string))
		return true
	})
	return ids
}

func (s *Server) updateClientConnections(clientID string, delta int32) {
	for {
		val, _ := s.activeConnections.LoadOrStore(clientID, uint32(0))
		current := val.(uint32)

		// 确保不会出现负数
		if delta < 0 && current == 0 {
			return
		}

		newCount := uint32(int32(current) + delta)
		if s.activeConnections.CompareAndSwap(clientID, current, newCount) {
			if delta > 0 {
				fmt.Printf("[连接] 客户端 %s 连接数更新: %d -> %d\n", clientID, current, newCount) // 添加控制台输出
			}
			return
		}
	}
}

func (s *Server) cleanupClient(clientID string) {
	// 清理客户端相关资源
	s.clients.Delete(clientID)
	s.lastHeartbeats.Delete(clientID)
	s.activeClients.Delete(clientID)
	s.updateClientState(clientID, false)
}

func (h *ClientHandler) forwardRequest(req *common.HTTPRequest) (*common.HTTPResponse, error) {
	fmt.Printf("[请求] 开始转发请求: %s %s\n", req.Method, req.Path)

	reqBytes, err := json.Marshal(req)
	if err != nil {
		return nil, fmt.Errorf("序列化请求失败: %v", err)
	}

	requestID := fmt.Sprintf("%d", time.Now().UnixNano())
	respChan := make(chan *common.TunnelMessage, 1)

	// 注册响应通道
	h.respMux.Lock()
	h.respChans[requestID] = respChan
	h.respMux.Unlock()

	// 确保清理响应通道
	defer func() {
		h.respMux.Lock()
		delete(h.respChans, requestID)
		h.respMux.Unlock()
		close(respChan)
	}()

	// 发送请求
	err = h.tunnel.SendMessage(&common.TunnelMessage{
		Type:      common.TypeHTTPRequest,
		RequestID: requestID,
		Payload:   reqBytes,
	})
	if err != nil {
		return nil, fmt.Errorf("发送请求失败: %v", err)
	}

	fmt.Printf("[请求] 已发送请求，等待响应 (ID: %s)\n", requestID)

	// 等待响应
	select {
	case resp := <-respChan:
		if resp == nil {
			return nil, fmt.Errorf("响应通道已关闭")
		}

		var httpResp common.HTTPResponse
		if err := json.Unmarshal(resp.Payload, &httpResp); err != nil {
			return nil, fmt.Errorf("解析响应失败: %v", err)
		}

		fmt.Printf("[响应] 收到响应 (ID: %s, 状态码: %d)\n", requestID, httpResp.StatusCode)
		return &httpResp, nil

	case <-time.After(30 * time.Second):
		return nil, fmt.Errorf("请求超时")
	}
}

func (h *ClientHandler) handleResponse(msg *common.TunnelMessage) {
	h.respMux.RLock()
	respChan, exists := h.respChans[msg.RequestID]
	h.respMux.RUnlock()

	if exists {
		respChan <- msg
	} else {
		fmt.Printf("[警告] 收到未知请求ID的响应: %s\n", msg.RequestID)
	}
}
